package com.atguigu.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object CategoryTop10_ACC {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    conf .set("spark.testing.memory","21474800000")
    val sc: SparkContext = new SparkContext(conf)
    val log: RDD[String] = sc.textFile("spark-core/data/user_visit_action.txt")
    val acc: CategoryAcc = new CategoryAcc
    sc.register(acc,"CategoryACC")
    log.foreach(log=>{
      val strs: Array[String] = log.split("_")
      if (strs(6) != "-1") {
        acc.add((strs(6),"click"))
      }
      else if (strs(8) != "null") {
        strs(8).split(",").foreach(x=>{
          acc.add((x,"order"))
        })
      }
      else if (strs(10) != "null") {
        strs(10).split(",").foreach(x=>{
          acc.add((x,"pay"))
        })      }
    })
    println(acc.value.toList.sortBy(x=>{
      (x._2.click,x._2.order,x._2.pay)
    })(Ordering.Tuple3(Ordering.Long.reverse,Ordering.Long.reverse,Ordering.Long.reverse)).take(10))
    sc.stop()
  }
}

case class Category(){
  var click:Long = 0
  var order:Long = 0
  var pay:Long = 0

  override def toString: String = {
    s"($click , $order , $pay)"
  }
}

class CategoryAcc extends AccumulatorV2[(String,String),mutable.Map[String,Category]]{
  var map:mutable.Map[String,Category] = mutable.Map()
  override def isZero: Boolean = map.isEmpty

  override def copy(): AccumulatorV2[(String, String), mutable.Map[String, Category]] = {new CategoryAcc}

  override def reset(): Unit = map.clear()

  override def add(v: (String, String)): Unit = {
    val category: Category = map.getOrElse(v._1, new Category)
    v._2 match {
      case "click" => category.click+=1
      case "order" => category.order+=1
      case "pay" => category.pay+=1
    }
    map.update(v._1,category)
  }

  override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, Category]]): Unit = {
    val otherMap: mutable.Map[String, Category] = other.value
    map = this.map.foldLeft(otherMap)(
      (innerMap,kv)=>{
        val category: Category = innerMap.getOrElse(kv._1, new Category)
        category.click=kv._2.click
        category.order=kv._2.order
        category.pay=kv._2.pay
        innerMap.update(kv._1,category)
        innerMap
      }
    )
  }

  override def value: mutable.Map[String, Category] = {
    map
  }
}
